# -*- coding:utf8 -*-
# @Time   : 2021/7/8 16:02
# @Author : tianyv
# @File   : demo.py
# @Content:看板指标模板

import datetime
from com.lcb.bigdata.util import ApolloUtil, DBUtil, KafkaUtil, DateUtil
from com.lcb.bigdata.job.BaseJob import BaseJob
from com.lcb.bigdata.util.LogUtil import LogUtil

logger = LogUtil("mars-job-mic-indicator", __name__)

class Job(BaseJob):
    def __init__(self, appId=None):
        """
        :param appId: 需要指定apollo中配置的的appId
        """
        super(Job, self).__init__(appId)
        self.__name__ = '${JOB_NAME}'

    def doInitJob(self):
        """
        初始化JOB
        :param appId:
        """
        print(self.paras)

        ##  填写指标编码
        self.indicator_code = "${INDICATOR_CODE}"

        # 通过Apollo获取kafka配置
        kafka_bootstrap_servers = ApolloUtil.getApolloValueByKey(self.apollo_client, 'dataplatform_kafka_bootstrap_servers')
        self.producer = KafkaUtil.get_producer(kafka_bootstrap_servers)

        # 通过Apollo获取kudu配置
        kuduHost = ApolloUtil.getApolloValueByKey(self.apollo_client, 'mars_kudu_maste_address')
        kuduDatabase = ApolloUtil.getApolloValueByKey(self.apollo_client, 'mars_kudu_database')
        self.conn_impl = DBUtil.connectImpalaDB(kuduDatabase, kuduHost)

    def insert_kudu(self,indicator_date,indicator_cycle,start_timestamp,end_timestamp):
        print("===============开始计算并入kudu库[{0}]".format(self.indicator_code))
        logger.info("===============开始计算并入kudu库[{0}]".format(self.indicator_code))
        ##todo 填写upsert语句
        sql = '''
            ${INSERT_KUDU_SQL}
        '''.format(self.indicator_code,indicator_date,indicator_cycle,start_timestamp,end_timestamp)
        print(sql)
        DBUtil.exeImpalaSql2(sql, self.conn_impl)


    def kudu_to_kafka(self,indicator_date,indicator_cycle):
        print("===============开始查询并发往kafka[{0}]".format(self.indicator_code))
        logger.info("===============开始查询并发往kafka[{0}]".format(self.indicator_code))

        ## 查询sql
        sql = '''
            ${KUDU_TO_KAFKA_SQL}
        '''.format(self.indicator_code,indicator_date,indicator_cycle)

        dataList = DBUtil.exeSelectSqlWithSchema(sql,self.conn_impl)

        ## 填写es的index
        KafkaUtil.batch_send_kudu2es("ads_kudu2es", self.producer, dataList, index="${ES_INDEX}")


    def doProcess(self):
        """
        执行业务代码
        :return:
        """
        logger.info("====>process data {}".format(self.__name__))
        startTime = datetime.datetime.strptime(self.paras['startTime'], '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%d')
        endTime = datetime.datetime.strptime(self.paras['endTime'], '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%d')
        indicator_cycle = self.paras['indicator_cycle']
        while startTime < endTime:
            start_timestamp, end_timestamp, indicator_date = DateUtil.isTimestampTypeBetween(startTime, indicator_cycle)

            ## 检查时间范围 (如近30天的指标 需重新设置时间范围)
            print("============计算时间范围从{0}到{1}".format(start_timestamp, end_timestamp))
            print("indicator_date = {0},indicator_cycle={1}".format(indicator_date,indicator_cycle))

            self.insert_kudu(indicator_date, indicator_cycle,start_timestamp,end_timestamp)
            self.kudu_to_kafka(indicator_date, indicator_cycle)

            startTime = (datetime.datetime.strptime(startTime, '%Y-%m-%d') + datetime.timedelta(days=1)).strftime('%Y-%m-%d')

    def doDestroy(self):
        """
        释放资源
        :return:
        """
        logger.info("====>destroy data {}".format(self.__name__))
        self.conn_impl.close()
        self.producer.close()

    def getStartTime(self) -> str:
        """
        如果需要动态设定开始时间，实现此方法
        :return:
        """
        return ${GET_START_TIME}

    def getEndTime(self) -> str:
        """
        如果需要动态设定结束时间，实现此方法
        :return:
        """
        return ${GET_END_TIME}


if __name__ == '__main__':
    Job('0030040011').proccess()